#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__
#include "buffer.hpp"
#include <mutex>
#include <condition_variable>
#include <functional>
#include <memory>
#include <atomic>
#include <thread>

namespace lyzlog
{
    enum class AsyncType
    {
        ASYNC_SAFE,
        ASYNC_UNSAFE
    };
    using Functor = std::function<void(Buffer &)>;
    class AsyncLooper
    {
    public:
        AsyncLooper(const Functor &cb, AsyncType looper_type = AsyncType::ASYNC_SAFE) : _stop(false), _looper_type(looper_type), _callback(cb), _thread(std::thread(&AsyncLooper::threadEntry, this))
        {
        }
        ~AsyncLooper()
        {
            stop();
            if (_thread.joinable())
            {
                _thread.join();
            }
        }
        using ptr = std::shared_ptr<AsyncLooper>;
        void stop()
        {
            _stop = true;
            _con_cond.notify_all();
        }
        void push(const char *data, int len)
        {
            // 1.无限扩容 -- 不安全（压力测试）2.固定大小--生产缓冲区中满了就会阻塞
            std::unique_lock<std::mutex> lock(_mutex);
            // 条件变量控制，若缓冲区剩余大小大于数据长度，才可以添加数据
            if (_looper_type == AsyncType::ASYNC_SAFE)
            {
                _pro_cond.wait(lock, [&]()
                               { return _pro_buff.writeAbleSize() >= len; });
            }
            // 走到这里说明满足条件，可以向缓冲区添加数据
            _pro_buff.push(data, len);
            // 唤醒消费者对缓冲区数据进行处理
            _con_cond.notify_one();
        }

    private:
        void threadEntry()
        {
            while (1)
            {
                // 1.判断生产缓冲区有无数据，有数据则交换，没有数据则阻塞
                // 为互斥锁设置生命周期，点那个缓冲区交换完毕后就解锁（并不对数据写入处理加锁）
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    //退出标记被设置，并且生产缓冲区无数据，才能退出
                    if (_stop && _pro_buff.empty())
                        break;
                    // 1.退出唤醒 2.当前生产者缓冲区不为空
                    _con_cond.wait(lock, [&]()
                                   { return _stop || !_pro_buff.empty(); });
                    _con_buff.swap(_pro_buff);
                    // 唤醒消费者
                    if (_looper_type == AsyncType::ASYNC_SAFE)
                        _pro_cond.notify_all();
                }
                // 2.被唤醒之后，对消费缓冲区数据进行处理
                _callback(_con_buff);
                // 3.初始化消费缓冲区`
                _con_buff.reset();
            }
        }

    private:
        Functor _callback;

    private:
        AsyncType _looper_type;
        std::atomic<bool> _stop;
        Buffer _pro_buff;                  // 生产者缓冲区
        Buffer _con_buff;                  // 消费者缓冲区
        std::mutex _mutex;                 // 互斥锁
        std::condition_variable _pro_cond; // 生产者条件变量
        std::condition_variable _con_cond; // 消费者条件变量
        std::thread _thread;
    };
}

#endif